package com.hns.tool.task.taque;

import java.util.HashSet;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 任务执行队列
 * @author:wucx
 * @email:zhuangcaijin@126.com
 * @version Revision 2.0.0
 */
public class Taque {

	private final Logger logger = LoggerFactory.getLogger(Taque.class);
	
	/**
	 * 待办任务列表
	 */
	private Queue<Task> todos = new PriorityBlockingQueue<Task>();
	/**
	 * 检查任务的唯一性，过滤重复的任务
	 */
	private Set<Task> unique = new HashSet<Task>();
	/**
	 * 任务提交者
	 */
	private TaskSubmitter submitter;
	/**
	 * 任务终结者
	 */
	private TaskTerminator terminator;
	/**
	 * 重入锁
	 */
	private ReentrantLock lock = new ReentrantLock();
	
	/**
	 * 执行中的任务
	 */
	Queue<Task> doings = new ConcurrentLinkedQueue<Task>();
	/**
	 * 执行中的任务数
	 */
	AtomicInteger length = new AtomicInteger(0);
	/**
	 * 最大执行中的任务数
	 */
	private int maxDoing = 2;
	/**
	 * 线程池
	 */
	ExecutorService pool = Executors.newCachedThreadPool();
	boolean isUp = true;
	
	/**
	 * 构造
	 * @param maxTaskInRun 最大任务数
	 * @email:zhuangcaijin@126.com
	 */
	public Taque(int maxTaskInRun) {
		this.maxDoing = maxTaskInRun;
		submitter = new TaskSubmitter(this);
		submitter.start();
		terminator = new TaskTerminator(this);
		terminator.start();
	}
	
	/**
	 * 关闭线程池
	 * @author:wucx
	 * @email:zhuangcaijin@126.com
	 */
	public void shutdown() {
		isUp = false;
		pool.shutdown(); // Disable new tasks from being submitted
		try {
		     // Wait a while for existing tasks to terminate
			if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
				pool.shutdownNow(); // Cancel currently executing tasks
				if (!pool.awaitTermination(60, TimeUnit.SECONDS))
					System.err.println("Pool did not terminate");
			}
		} catch (InterruptedException ie) {
			// (Re-)Cancel if current thread also interrupted
			pool.shutdownNow();
			// Preserve interrupt status
			Thread.currentThread().interrupt();
		}
	}
	
	/**
	 * 增加一个待办任务(加锁确保不会添加重复任务)
	 * @param task 后台任务
	 * @author:wucx
	 * @email:zhuangcaijin@126.com
	 */
	public void push(Task task) {
		// 检查唯一性
		boolean isAdded = false;
		lock.lock();
		try {
			isAdded = unique.add(task);
		} finally {
			lock.unlock();
		}
		if ( isAdded ) {
			todos.add(task);
			if ( submitter.getState() == Thread.State.TIMED_WAITING ) {
				// 唤醒提交者
				submitter.wakeup();
			}
			logger.debug("Task {} add to list size {} " , task , todos.size());
		} else {
			// 标记任务被忽略（已存在列表中）
			if ( ! todos.contains(task) && ! doings.contains(task) ) {
				task.state = Task.State.Ignore;
			}
		}
	}
	
	/**
	 * 执行并同步等待任务结束
	 * @param task 后台任务
	 * @return 任务结果
	 * @author:wucx
	 * @email:zhuangcaijin@126.com
	 */
	public Object runSync(final Task task) {
		doings.add(task);
		length.incrementAndGet();
		@SuppressWarnings("rawtypes")
		Future future = pool.submit(new TaskRun(task, this));
		try {
			future.get();
		} catch (Exception e) {
			logger.info("Task {} run error.", task);
		}
		return task.result;
	}

	/**
	 * 获取待办任务中第一个优先的任务
	 * 由提交者调用，以把任务加入到执行队列
	 * @return 加入到执行队列中的任务
	 * @author:wucx
	 * @email:zhuangcaijin@126.com
	 */
	Task take() {
		logger.trace("Before take todo list size {} " , todos.size());
		Task task = todos.poll();
		logger.trace("After take todo list size {} " , todos.size());
		if ( task != null ) {
			length.incrementAndGet();
			doings.add(task);
		}
		return task;
	}

	/**
	 * 任务执行完成后
	 * @param task 后台任务
	 * @author:wucx
	 * @email:zhuangcaijin@126.com
	 */
	void done(Task task) {
		doings.remove(task);
		length.decrementAndGet();
		logger.info("Task {} ran over, state:{}", task, task.state);
		// 触发任务完成的逻辑
		fireEnd(task);
		if ( submitter.getState() == Thread.State.TIMED_WAITING ) {
			submitter.wakeup();
		}
	}
	
	/**
	 * 任务结束时执行的动作
	 * @param task 后台任务
	 * @author:wucx
	 * @email:zhuangcaijin@126.com
	 */
	private void fireEnd(Task task) {
		TaskEnd te = new TaskEnd(task, this);
		execute(te);
	}
	
	/**
	 * 任务执行终止
	 * @param task 后台任务
	 * @author:wucx
	 * @email:zhuangcaijin@126.com
	 */
	void end(Task task) {
		lock.lock();
		try {
			unique.remove(task);
		} finally {
			lock.unlock();
		}
		logger.debug("Task {} finish.", task);
	}

	/**
	 * 执行是否繁忙
	 * @return 当前任务数是否超出了最大
	 * @author:wucx
	 * @email:zhuangcaijin@126.com
	 */
	boolean isBusy() {
		return length.get() >= maxDoing;
	}
	
	/**
	 * 执行后台任务
	 * @param run
	 * @author:wucx
	 * @email:zhuangcaijin@126.com
	 */
	void execute(Runnable run) {
		pool.submit(run);
	}

}
